package org.dt.common.sequence;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;
import org.dt.common.sequence.config.SeqHolderSpec;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;

import static java.lang.Math.max;

/**
 * 【Serial Number Generator】 Use {@link BlockingQueue} as the local cache
 *
 * <p>creation time: 2018/11/29 14:29</p>
 *
 * @author yujiadong
 * @version 0.2
 * @since 0.2
 * Changes：
 * 1）增加多种序列号请求量预估手段，补充号段更准确。包括：预设容量、被阻塞的序列号请求量、基于时间戳的滑动窗口流量统计
 * 2）将获取信号量的tryAcquire()操作前置到线程池submit之前，大量减少无效Task提交，降低线程池溢出风险和处理压力
 * 3）可以在while中处理线程中断了
 * 4）while如果需要异步补充号段，从轮询改为当前线程让步（yield）
 * 5）移除Spring相关依赖，将其转移到{@link cn.com.cathayins.fdt.common.sequence.configuration}
 * <p>
 * Issues：
 * Perf(i5 4cores,8G,4 threads):
 * - tps 100w/s
 * -
 * @since 0.1
 * Changes：
 * 1）所有请求只在一个可分配的Sequence上自旋，减少{@link GreedyPollSequenceHolder}中对cache队列的remove/insert操作
 * 2）拥塞控制：自适应调整step（步长），将补充号段缓存的数据库IO请求从 (capacity/step)次 减少到 1次
 * <p>
 * Issues：
 * 1.20: 冷启动+大流量场景下，耗时为平均时间的30-50倍(100ms)
 * 1.20: 存在周期性的毛刺，毛刺耗时为平均时间的10倍(20ms)
 **/
@SuppressWarnings("jol")
@Slf4j
public class HeadSpinningSequenceHolder extends AbstractSequenceHolder {
    private final static int                              OFFER_TIMEOUT = 3;
    /**
     * the ONLY variable where all requests try to get sequence from
     */
    private final        AtomicReference<TrackedSequence> current       = new AtomicReference<>(TrackedSequence.EMPTY);
    private final        float                            latestSlotRate;
    /**
     * semaphore to limit concurrent DB-accessing operations
     * always 1 because mysql can't update one row like Java CAS style
     */
    private final        Semaphore                        permits;
    /**
     * the cache pool
     */
    private final        BlockingQueue<TrackedSequence>   pool          = new LinkedBlockingQueue<>();
    private final        int                              statSlotSize;
    private volatile     long                             lastStep;
    private              AtomicLong                       requiredSeqWidth;
    /**
     * traffic statistic for sequence width required
     */
    private              Statistic                        stat;

    public HeadSpinningSequenceHolder(String biz, SeqHolderSpec config,
                                      BaseDelegator stepGenerator, Executor taskExecutor) {

        super(biz, config, stepGenerator, taskExecutor);
        this.permits = new Semaphore(1);

        this.lastStep         = this.capacity;
        this.requiredSeqWidth = new AtomicLong(capacity);

        this.statSlotSize   = config.getStatSlotSize();
        this.latestSlotRate = config.getLatestSlotRate();

        stat = new Statistic(statSlotSize, config.getMsEachStatSlot());
    }

    @Override
    public List<Sequence> doApplyFor(Long need) {
        stat.onTrafficIn(System.currentTimeMillis(), need);
        long remain = need;

        // lazy invoke on cache supplying
        long availableWidth = availableWidth();
        if (availableWidth - remain < lastStep * loadFactor) {
            log.debug("[{}] remaining {} with available width {}", biz, remain, availableWidth);
            asyncSupplySequence(remain, true);
        }

        boolean        addToRequired = true;
        List<Sequence> list          = new LinkedList<>();
        while (remain != 0 && !Thread.currentThread().isInterrupted()) {
            // when application loaded, default value of current is EMPTY
            TrackedSequence current = this.current.get();
            if (current.isEmpty()) {
                // try to fetch new sequence from cache if not empty
                TrackedSequence gotNew = null;
                try {
                    //为什么不直接使用pool.poll()？可以避免一秒的等待
                    //经过测试，发现Max latency没有质的提升（从200ms->190ms）
                    gotNew = pool.poll(1, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.warn("found Interrupted request that is not excepted, just ignore here", e);
                }

                // try to fetch new sequence from DB
                if (gotNew == null) {
                    log.debug("[{}] can't get new sequence from cache pool", biz);
                    asyncSupplySequence(remain, addToRequired);
                    addToRequired = false;

                    Thread.yield();
                }
                // try to replace current to new sequence got before,
                // fall back to poll if failed
                else if (this.current.compareAndSet(current, gotNew)) {
                    log.debug("[{}] replace current success with {}", biz, gotNew);

                } else {
                    try {
                        // todo how to determine timeout?
                        pool.offer(gotNew, OFFER_TIMEOUT, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        log.warn("found Interrupted request that is not excepted, just ignore here", e);
                    }
                }
                continue;
            }

            // first try consume the HEAD node
            Sequence sequence;
            if (Sequence.EMPTY != (sequence = current.tryConsume(remain))) {
                remain -= sequence.width();
                list.add(sequence);
            } else {
                log.debug("[{}] tryConsume() return EMPTY. remain {}", biz, remain);
            }
        }

        return list;
    }

    @Override
    public void init() {
        try {
            delegator.init(biz);

            doSupplySequence();

            if (!pool.isEmpty()) {
                current.set(pool.poll());
            }
            log.debug("[{}] init finished", biz);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error(e.getMessage(), e);
        }
    }

    private void asyncSupplySequence(long need, boolean addToRequired) {
        if (addToRequired) {
            increaseRequiredSeq(need);
        }
        if (permits.tryAcquire()) {
            taskExecutor.execute(() -> {
                try {
                    doSupplySequence();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error(e.getMessage(), e);
                } finally {
                    permits.release();
                }
            });
        }
    }

    private long availableWidth() {
        return Stream.concat(pool.stream(), Stream.of(current.get()))
                     .map(TrackedSequence::availableWidth).reduce(0L, Long::sum);
    }

    private long calcRequiredCountAndReset() {
        return requiredSeqWidth.getAndSet(0L);
    }

    private long calcStep() {
        long now = System.currentTimeMillis();

        double avg       = stat.avgTraffic(now, capacity / statSlotSize);
        double latestAvg = stat.latestAvgTraffic(now, capacity / statSlotSize, latestSlotRate);
        log.debug("get avg traffic {}, latest traffic avg {} from stat", avg, latestAvg);

        double nextTrafficRate = Math.max(avg, latestAvg);

        long step = max(max((long) (nextTrafficRate * statSlotSize), calcRequiredCountAndReset()), capacity);
        this.lastStep = step;
        return step;
    }

    private void doSupplySequence() throws InterruptedException {
        long s = System.currentTimeMillis();

        long     step     = calcStep();
        Sequence sequence = delegator.applyForOneStep(biz, step);
        //for test to mock real DB
        //Thread.sleep(100);

        Validate.notNull(sequence, "[{}] get new sequence failed, there must be something wrong", biz);
        pool.offer(new TrackedSequence(sequence.getLowerBound(), sequence.getUpperBound()), OFFER_TIMEOUT, TimeUnit.SECONDS);
        log.info("[{}] 已补充号段 {} , step {}, 耗时 {}", biz, sequence, step, (System.currentTimeMillis() - s));
    }

    private void increaseRequiredSeq(long need) {
        requiredSeqWidth.getAndAdd(need);
    }
}
